package com.sangsang.kafka.util;


import com.sangsang.kafka.domain.dto.PayloadDto;

import java.util.Comparator;
import java.util.List;
import java.util.stream.Collectors;

/**
 * debezium kafka的数据解析工具类
 *
 * @author liutangqi
 * @date 2023/12/25 10:22
 */
public class DataParseUtils {

    /**
     * 处理修改后的数据不为null的数据
     * 1.过滤去除 我们需要同步的字段中 变更前和变更后一样的数据，这部分数据对我们来说并没有变更
     * 2.根据主键去重，取binlog文件名最大，pos位置最大的数据，舍弃其它数据，类似于redis的AOF重写
     * 3.获取处理后的变更后的数据
     *
     * @author liutangqi
     * @date 2023/12/25 15:46
     * @Param [datas]
     **/
    public static <T> List<T> parse(List<PayloadDto<T>> datas) {
        return datas.stream()
                .filter(f -> f != null && f.getAfter() != null)
                .filter(f -> !CDCUtils.equal(f.getBefore(), f.getAfter()))
                .collect(Collectors.groupingBy(g -> CDCUtils.getPrimaryKeyValue(g.getAfter())))
                .entrySet()
                .stream()
                .map(m -> m.getValue().stream().sorted(Comparator.comparing(PayloadDto::getSource)).findFirst().orElse(null))
                .map(PayloadDto::getAfter)
                .collect(Collectors.toList());
    }

    /**
     * 解析删除了的数据的主键
     *
     * @author liutangqi
     * @date 2023/12/28 15:59
     * @Param [datas]
     **/
    public static <T> List parseDeleteIds(List<PayloadDto<T>> datas) {
        return datas.stream()
                .filter(f -> f != null && f.getAfter() == null && f.getBefore() != null)
                .map(PayloadDto::getBefore)
                .map(CDCUtils::getPrimaryKeyValue)
                .collect(Collectors.toList());
    }

}
